package com.kyp.basic;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.log4j.Logger;

import com.kyp.factory.KYPMapReduce;
import com.kyp.util.Validation;


/***************************************************
 * 
 * File Format
 * Group Key, Group Key, Value, Value, Value  
 * Group Key, Group Key, Value, Value, Value
 * Group Key, Group Key, Value, Value, Value
 * 
 * EX)
 * A,K,123,234,122
 * A,K,111,222,333
 * B,K,333,334,322
 * B,K,341,123,532
 * 
 */


public class BasicCalc extends KYPMapReduce {

	private static final Logger sLogger = Logger.getLogger(BasicCalc.class);
	
	private final String APP_NAME = "Basic Calculation";
	
	public final static String RESULT_FILE_NAME = "bascalc";
	
	
	@Override
	protected void execute(Configuration conf, String[] otherArgs) {
		String groupCount = conf.get("group"); 		
		if(groupCount == null || groupCount == ""){
			sLogger.info("Groupping 하고싶은 Left Column Count가 있어야 합니다.");
			return;
		}		
		
		if(!Validation.isInteger(groupCount)){
			sLogger.info("Group Count는 숫자이어야 합니다.[" + groupCount + "]");
			return;			
		}
		
		try {
			//Job Name 설정
			Job job = new Job(conf, APP_NAME);
			
			//입출력 데이터 경로 설정
			FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
			FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
			
			//Job 클래스 설정 
			job.setJarByClass(BasicCalc.class);
			
			job.setPartitionerClass(BasicCalcGroupKeyPartitioner.class);
			job.setGroupingComparatorClass(BasicCalcKeyComparator.class);
			job.setSortComparatorClass(BasicCalcKeyComparator.class);
			
			//Mapper 클래스 설정
			job.setMapperClass(BasicCalcMapper.class);
			//Reducer 클래스 설정 
			job.setReducerClass(BasicCalcReducer.class);
			
			job.setMapOutputKeyClass(BasicCalcKey.class);
			job.setMapOutputValueClass(DoubleWritable.class);
			
			//입출력 데이터 포맷 설정
			job.setInputFormatClass(TextInputFormat.class);
			job.setOutputFormatClass(TextOutputFormat.class);
			
			job.setOutputKeyClass(Text.class);
			job.setOutputValueClass(DoubleWritable.class);
			
			MultipleOutputs.addNamedOutput(job, RESULT_FILE_NAME, TextOutputFormat.class, Text.class, DoubleWritable.class);
			
			job.waitForCompletion(true);
		} catch (IOException ioe) {
			sLogger.error(ioe.getMessage(), ioe);
		} catch (ClassNotFoundException cne){
			sLogger.error(cne.getMessage(), cne);
		} catch (InterruptedException ite) {
			sLogger.error(ite.getMessage(), ite);
		}
	}
	
	
}
